package zhang.app.dws;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import zhang.app.BaseAppV1;
import zhang.bean.TestQuestionBean;
import zhang.common.Constant;
import zhang.util.AtguiguUtil;
import zhang.util.FlinkSinkUtil;

import java.math.BigInteger;
import java.time.Duration;
import java.util.HashMap;


public class DwsTestQuestionWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new DwsTestQuestionWindow()
                .init(3014,2,"DwsTestQuestionWindow", Constant.TOPIC_DWD_TEST_QUESTION);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1. 封装到bean
        SingleOutputStreamOperator<TestQuestionBean> beanStream = pasebean1(stream);

        // 2. 开窗聚合
        SingleOutputStreamOperator<TestQuestionBean> resultStream = windowAndAgg1(beanStream);


        // 3. 写出
        writeToDoris(resultStream);
    }

    private void writeToDoris(SingleOutputStreamOperator<TestQuestionBean> resultStream) {
        resultStream.map(bean -> {
            SerializeConfig config = new SerializeConfig();
            config.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
            String json = JSON.toJSONString(bean, config);
            System.out.println(json);
            return json;
        })
                .addSink(FlinkSinkUtil.getDorisSink("dws_db.dws_test_question_id_view_window"));
    }

    private SingleOutputStreamOperator windowAndAgg1(SingleOutputStreamOperator<TestQuestionBean> beanStream) {
    return     beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.
                <TestQuestionBean>forBoundedOutOfOrderness(Duration.ofSeconds(60))
                .withTimestampAssigner((bean,ts)->bean.getTs()*1000)
        )
                .keyBy(bean->bean.getQuestionId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
               .reduce(new ReduceFunction<TestQuestionBean>() {

                           @Override
                public TestQuestionBean reduce(TestQuestionBean bean1, TestQuestionBean bean2) throws Exception {
                               HashMap<String,String> userid= new HashMap<String,String>();
                               HashMap<String,String> replyuserid= new HashMap<String,String>();

                              if (bean2.getIsCorrect().equals("1")){
                                  bean1.setRigntCount(bean1.getRigntCount()+1L);
                                  if (userid.isEmpty()){
                                      bean1.setRigntUserCount(bean1.getRigntUserCount()+1L);
                                      userid.put(bean2.getUserId(),"");
                                  }else if (!userid.containsKey(bean2.getUserId())){
                                      bean1.setRigntUserCount(bean1.getRigntUserCount()+1L);
                                      userid.put(bean2.getUserId(),"");
                                  }
                              }
                               if (replyuserid.isEmpty()){
                                   bean1.setReplyUserCount(bean1.getReplyUserCount()+1L);
                                   replyuserid.put(bean2.getUserId(),"");
                               }else if (!replyuserid.containsKey(bean2.getUserId())){
                                   bean1.setReplyUserCount(bean1.getReplyUserCount()+1L);
                                   replyuserid.put(bean2.getUserId(),"");
                               }

                              bean1.setReplyCount(bean1.getReplyCount()+ bean2.getReplyCount());
                               return  bean1;
                           }
                       },
                       new ProcessWindowFunction<TestQuestionBean, TestQuestionBean, String, TimeWindow>() {
                           @Override
                           public void process(String s,
                                               Context context,
                                               Iterable<TestQuestionBean> elements,
                                               Collector<TestQuestionBean> out) throws Exception {
                               TestQuestionBean bean = elements.iterator().next();
                               bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                               bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                               bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                               out.collect(bean);
                           }
                       });



    }

    private SingleOutputStreamOperator pasebean1(DataStreamSource<String> stream) {
        return   stream.map(json-> {
            JSONObject object = JSON.parseObject(json);
            return   new TestQuestionBean(
                    "",
                    "",
                    object.getString("question_id"),
                    object.getString("user_id"),
                    object.getString("is_correct"),
                    0L,
                    1L,
                    0L,
                    0L,
                    "",
                    object.getLong("ts")
            );

        });
    }
}
